/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.LinkedList;
import java.util.List;
import java.util.ListIterator;
import java.util.Set;

import org.apache.zookeeper.ClientCnxn.EndOfStreamException;
import org.apache.zookeeper.ClientCnxn.Packet;
import org.apache.zookeeper.ZooDefs.OpCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ClientCnxnSocketNIO extends ClientCnxnSocket {
    private static final Logger LOG = LoggerFactory
            .getLogger(ClientCnxnSocketNIO.class);

    private final Selector selector = Selector.open();

    private SelectionKey sockKey;

    ClientCnxnSocketNIO() throws IOException {
        super();
    }

    /**
     * 只是说SelectionKey有没有初始化， 并不是真正的Connected
     * @return
     */
    @Override
    boolean isConnected() {
        return sockKey != null;
    }
    
    /**
     * 读就绪时,读取response
     * 写就绪时,从findSendablePacket找到可发送的Packet
     *
     *   主要分为读或者写两个case
     *   读:
     *     没有初始化就完成初始化
     *     读取len再给incomingBuffer分配对应空间
     *     读取对应的response
     *   写:
     *     找到可以发送的Packet
     *     如果Packet的byteBuffer没有创建，那么就创建
     *     byteBuffer写入socketChannel
     *     把Packet从outgoingQueue中取出来，放到pendingQueue中
     *     相关读写的处理
     *
     * 注意，读的时候是分两次读的：唯一能够想到的优点就是节省空间了。
     *     第1次只读len，然后给incomingBuffer分配对应的空间
     *     第2次再把剩下的内容读完
     * @return true if a packet was received
     * @throws InterruptedException
     * @throws IOException
     */
    void doIO(List<Packet> pendingQueue, LinkedList<Packet> outgoingQueue, ClientCnxn cnxn) throws InterruptedException, IOException {
        SocketChannel sock = (SocketChannel) sockKey.channel();
        if (sock == null) {
            throw new IOException("Socket is null!");
        }

        // 若读就绪
        if (sockKey.isReadable()) {
            int rc = sock.read(incomingBuffer);  // 读出len
            // 如果<0,表示读到末尾了,这种情况出现在连接关闭的时候
            if (rc < 0) {
                throw new EndOfStreamException( "Unable to read additional data from server sessionid 0x" + Long.toHexString(sessionId) + ", likely server has closed socket");
            }
            // 如果还有数据
            if (!incomingBuffer.hasRemaining()) {
                incomingBuffer.flip();  // 切换到读模式
                if (incomingBuffer == lenBuffer) {
                    recvCount++;    // 接收次数+1
                    readLength();   // 获取len并给incomingBuffer分配对应空间
                }
                // 如果client和server的连接还没有初始化
                else if (!initialized) {
                    // 第一次连接调到此处
                    readConnectResult();    // 读取connect 回复
                    enableRead();           // 启用读
                    // 如果有可以发送的packet
                    if (findSendablePacket(outgoingQueue, cnxn.sendThread.clientTunneledAuthenticationInProgress()) != null) {
                        enableWrite(); // 允许写，因为有要发送的packet
                    }
                    lenBuffer.clear();
                    incomingBuffer = lenBuffer; // 还原incomingBuffer
                    updateLastHeard();
                    initialized = true; // client和server连接初始化完成
                }
                // 如果已连接，并且已经给incomingBuffer分配了对应len的空间
                else {
                    sendThread.readResponse(incomingBuffer); // 读取response
                    lenBuffer.clear();
                    incomingBuffer = lenBuffer; // 还原incomingBuffer
                    updateLastHeard();
                }
            }
        }
        // 若写就绪
        if (sockKey.isWritable()) {
            synchronized(outgoingQueue) {
                // 找到可以发送的Packet, 这里并没有从队列头部删除掉
                Packet p = findSendablePacket(outgoingQueue, cnxn.sendThread.clientTunneledAuthenticationInProgress());

                if (p != null) {
                    updateLastSend();
                    if (p.bb == null) {
                        if ((p.requestHeader != null) &&
                                        (p.requestHeader.getType() != OpCode.ping) &&
                                        (p.requestHeader.getType() != OpCode.auth)) {
                            p.requestHeader.setXid(cnxn.getXid());
                        }
                        // 如果packet还没有生成byteBuffer，那就生成byteBuffer
                        p.createBB();
                    }
                    // 把packet的数据采用ByteBuffer的模式通过socket写出去
                    sock.write(p.bb); // 直接把请求（统一封装，类型、数据等等）
                    // 这个代码是处理拆包的，你发送数据的时候，就怕一次性没有发送完全
                    if (!p.bb.hasRemaining()) {
                        sentCount++;
                        // 把它从队列头部删除掉
                        outgoingQueue.removeFirstOccurrence(p);
                        if (p.requestHeader != null
                                        && p.requestHeader.getType() != OpCode.ping
                                        && p.requestHeader.getType() != OpCode.auth) {
                            synchronized (pendingQueue) {
                                // 加入待回复的队列, 此时需要等待这个请求的响应
                                pendingQueue.add(p);
                            }
                        }
                    }
                }
                if (outgoingQueue.isEmpty()) {
                    // 如果发现outgoingQueue是空的，此时就取消socket的OP_WRITE事件的关注
                    disableWrite();
                } else { //其他情况将OP_WRITE开启
                    enableWrite();
                }
            }
        }
    }

    /**
     * 根据sasl以及outgoingQueue情况获取发送的Packet
     * 主要流程：
     *     如果没有要发送的就返回null
     *     如果有要发送的或者client没有在处理sasl的权限，那么就拿队列第一个
     *     如果在处理sasl，那么遍历队列，把没有requestHeader为null的放到队头，返回该packet
     * @param outgoingQueue
     * @param clientTunneledAuthenticationInProgress  如果当前client和server在处理sasl的权限
     * @return
     */
    private Packet findSendablePacket(LinkedList<Packet> outgoingQueue, boolean clientTunneledAuthenticationInProgress) {
        synchronized (outgoingQueue) {
            // 如果没有要发送的
            if (outgoingQueue.isEmpty()) {
                return null;
            }
            // 如果有要发送的 或者 没有在处理sasl的权限
            if (outgoingQueue.getFirst().bb != null   || !clientTunneledAuthenticationInProgress) {
                return outgoingQueue.getFirst();
            }

            ListIterator<Packet> iter = outgoingQueue.listIterator();
            while (iter.hasNext()) {
                Packet p = iter.next();
                // 如果在处理sasl的权限，那么只有requestHeader为null的Packet可以被发送
                if (p.requestHeader == null) {
                    iter.remove();
                    outgoingQueue.add(0, p);
                    return p;
                } else {
                    if (LOG.isDebugEnabled()) {
                        LOG.debug("deferring non-priming packet: " + p + "until SASL authentication completes.");
                    }
                }
            }
            return null;
        }
    }

    @Override
    void cleanup() {
        // 此时就直接认为这个网络连接出现了问题
        if (sockKey != null) {
            SocketChannel sock = (SocketChannel) sockKey.channel();
            sockKey.cancel();
            try {
                sock.socket().shutdownInput();
            } catch (IOException e) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Ignoring exception during shutdown input", e);
                }
            }
            try {
                sock.socket().shutdownOutput();
            } catch (IOException e) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Ignoring exception during shutdown output", e);
                }
            }
            try {
                sock.socket().close();
            } catch (IOException e) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Ignoring exception during socket close", e);
                }
            }
            try {
                sock.close();
            } catch (IOException e) {
                if (LOG.isDebugEnabled()) {
                    LOG.debug("Ignoring exception during channel close", e);
                }
            }
        }
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            if (LOG.isDebugEnabled()) {
                LOG.debug("SendThread interrupted during sleep, ignoring");
            }
        }
        sockKey = null;
    }
 
    @Override
    void close() {
        try {
            if (LOG.isTraceEnabled()) {
                LOG.trace("Doing client selector close");
            }
            selector.close();
            if (LOG.isTraceEnabled()) {
                LOG.trace("Closed client selector");
            }
        } catch (IOException e) {
            LOG.warn("Ignoring exception during selector close", e);
        }
    }
    
    /**
     * 创建SocketChannel
     * @return the created socket channel
     * @throws IOException
     */
    SocketChannel createSock() throws IOException {
        SocketChannel sock;
        sock = SocketChannel.open();
        // 非阻塞
        sock.configureBlocking(false);
        sock.socket().setSoLinger(false, -1);
        sock.socket().setTcpNoDelay(true);
        return sock;
    }

    /**
     * register with the selection and connect
     * @param sock the {@link SocketChannel} 
     * @param addr the address of remote host
     * @throws IOException
     */
    void registerAndConnect(SocketChannel sock, InetSocketAddress addr)  throws IOException {
        // 注册，监听connect事件
        sockKey = sock.register(selector, SelectionKey.OP_CONNECT);
        // 因为SocketChannel sock是非阻塞的，此时并没有马上建立真正的连接就会返回true
        boolean immediateConnect = sock.connect(addr);
        if (immediateConnect) {
            // client把watches和authData等数据发过去，并更新SelectionKey为读写
            sendThread.primeConnection();
        }
    }

    /**
     *
     * @param addr  某一个zk server的地址
     * @throws IOException
     */
    @Override
    void connect(InetSocketAddress addr) throws IOException {
        SocketChannel sock = createSock();
        try {
            // 注册SelectionKey到zk server
           registerAndConnect(sock, addr);
        } catch (IOException e) {
            LOG.error("Unable to open socket to " + addr);
            sock.close();
            throw e;
        }
        // 还没有初始化,connect ok了但是还读到server的response
        initialized = false;

        lenBuffer.clear();
        incomingBuffer = lenBuffer;
    }

    /**
     * Returns the address to which the socket is connected.
     * 
     * @return ip address of the remote side of the connection or null if not
     *         connected
     */
    @Override
    SocketAddress getRemoteSocketAddress() {
        // a lot could go wrong here, so rather than put in a bunch of code
        // to check for nulls all down the chain let's do it the simple
        // yet bulletproof way
        try {
            return ((SocketChannel) sockKey.channel()).socket()
                    .getRemoteSocketAddress();
        } catch (NullPointerException e) {
            return null;
        }
    }

    /**
     * Returns the local address to which the socket is bound.
     * 
     * @return ip address of the remote side of the connection or null if not
     *         connected
     */
    @Override
    SocketAddress getLocalSocketAddress() {
        // a lot could go wrong here, so rather than put in a bunch of code
        // to check for nulls all down the chain let's do it the simple
        // yet bulletproof way
        try {
            return ((SocketChannel) sockKey.channel()).socket()
                    .getLocalSocketAddress();
        } catch (NullPointerException e) {
            return null;
        }
    }

    @Override
    synchronized void wakeupCnxn() {
        selector.wakeup();
    }

    /**
     * 如果 连接就绪，调用sendThread连接操作
     * 若 读写就绪，调用doIO函数
     * @param waitTimeOut
     * @param pendingQueue
     * @param outgoingQueue
     * @param cnxn
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    void doTransport(int waitTimeOut, List<Packet> pendingQueue, LinkedList<Packet> outgoingQueue, ClientCnxn cnxn) throws IOException, InterruptedException {
        // 之前基于NIO的连接建立完毕之后，它关注的是OP_READ 和 OP_WRITE
        // 当前这个连接如果有zk服务器的响应发送回来，此时会触发OP_READ, 如果是可以写数据，会触发OP_WRITE
        selector.select(waitTimeOut);

        Set<SelectionKey> selected;
        synchronized (this) {
            selected = selector.selectedKeys();
        }
        updateNow();

        for (SelectionKey k : selected) {
            SocketChannel sc = ((SocketChannel) k.channel());
            // 如果就绪的是connect事件，这个出现在registerAndConnect函数没有立即连接成功
            if ((k.readyOps() & SelectionKey.OP_CONNECT) != 0) {
                // 如果完成了连接
                if (sc.finishConnect()) {
                    // 更新时间
                    updateLastSendAndHeard();
                    // client把watches和authData等数据发过去，并更新SelectionKey为读写
                    sendThread.primeConnection();
                }
            }
            // 如果就绪的是读或者写事件
            else if ((k.readyOps() & (SelectionKey.OP_READ | SelectionKey.OP_WRITE)) != 0) {
                // 核心代码，  如果有读写事件可以执行，此时就是在这里去执行对应的网络IO操作
                // 利用pendingQueue和outgoingQueue进行IO,
                doIO(pendingQueue, outgoingQueue, cnxn);
            }
        }
        // 如果zk的state是已连接
        if (sendThread.getZkState().isConnected()) {
            synchronized(outgoingQueue) { //没有消息会取消wirte操作,在有可写内容时再注册write操作
                // 如果有可以发送的packet
                if (findSendablePacket(outgoingQueue, cnxn.sendThread.clientTunneledAuthenticationInProgress()) != null) {
                    // 允许写
                    enableWrite();
                }
            }
        }
        // 清空
        selected.clear();
    }

    @Override
    void testableCloseSocket() throws IOException {
        LOG.info("testableCloseSocket() called");
        ((SocketChannel) sockKey.channel()).socket().close();
    }

    @Override
    synchronized void enableWrite() {
        int i = sockKey.interestOps();
        if ((i & SelectionKey.OP_WRITE) == 0) {
            sockKey.interestOps(i | SelectionKey.OP_WRITE);
        }
    }

    @Override
    public synchronized void disableWrite() {
        int i = sockKey.interestOps();
        if ((i & SelectionKey.OP_WRITE) != 0) {
            sockKey.interestOps(i & (~SelectionKey.OP_WRITE));
        }
    }

    synchronized private void enableRead() {
        int i = sockKey.interestOps();
        if ((i & SelectionKey.OP_READ) == 0) {
            sockKey.interestOps(i | SelectionKey.OP_READ);
        }
    }

    @Override
    synchronized void enableReadWriteOnly() {
        // 后续之监听 读写请求
        sockKey.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
    }

    Selector getSelector() {
        return selector;
    }

    @Override
    void sendPacket(Packet p) throws IOException {
        SocketChannel sock = (SocketChannel) sockKey.channel();
        if (sock == null) {
            throw new IOException("Socket is null!");
        }
        p.createBB();
        ByteBuffer pbb = p.bb;
        sock.write(pbb);
    }


}
